package com.sam.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @author: wm.xue
 * @date: 2021/11/4 17:06
 * @description:
 */
public class _02_MyConsumer {

    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.100.17.108:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "_02_myConsumer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);
        // 订阅分区 正则
        Pattern pattern = Pattern.compile("tp_[0-9]_demo");
        consumer.subscribe(pattern);

//        consumer.subscribe(pattern, new ConsumerRebalanceListener() {
//            @Override
//            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
//            }
//            @Override
//            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
//            }
//        });

        // 订阅主题分区
        List<String> topics = Arrays.asList("tp_01_demo", "tp_02_demo");
        consumer.subscribe(topics);

        consumer.subscribe(topics, new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {

            }
        });
        // 拉取消息
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));

        // 获取tp_01_demo主题消息
        Iterable<ConsumerRecord<String, String>> topic = records.records("tp_01_demo");
        topic.forEach(record -> {
            record.offset();
            record.topic();
        });



    }

}
